package com.tmsb.mall;

import com.alibaba.fastjson.JSON;
import com.tmsb.pojo.MysqlBinlog;
import com.tmsb.pojo.OrderGoods;
import com.tmsb.utils.*;
import com.tmsb.utils.Const.MysqlDatabase;
import com.tmsb.utils.Const.MysqlTable;
import com.tmsb.utils.trigger.TimerTrigger;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.JedisCluster;
import scala.Tuple4;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Description: 滚动每日营业额，订单数，另外清单显示排名前十的商家,商品topicN
 * Version:1.0.0
 *
 * @author shihb
 * @date 2020/6/10 16:47
 */
public class OrderAnalysis {

  private static OutputTag<OrderGoods> orderGoodTag = new OutputTag<OrderGoods>("orderGoodsTag") {
  };

  private static Logger LOG = LoggerFactory.getLogger(OrderAnalysis.class);

  public static void main(String[] args) {
    try {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
      env.enableCheckpointing(ConfUtil.getCheckpointInterval(), CheckpointingMode.EXACTLY_ONCE);
      env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
      env.setStateBackend(new FsStateBackend(ConfUtil.getCheckpointUri()));
//      env.setStateBackend(new FsStateBackend("file:///C:\\Users\\home\\Desktop\\mall\\orderGoods"));
      env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
      env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
      env.getCheckpointConfig().setMinPauseBetweenCheckpoints(ConfUtil.getCheckpointMinPauseBetween());
      env.setParallelism(ConfUtil.getParallelism());

      env.setRestartStrategy(RestartStrategies.failureRateRestart(
              ConfUtil.getfailureRate(), org.apache.flink.api.common.time.Time.of(ConfUtil.getfailureInterval(), TimeUnit.MINUTES),
              org.apache.flink.api.common.time.Time.of(ConfUtil.getdelayInterval(), TimeUnit.SECONDS)
      ));



      DataStream<String> source = env
          .addSource(
              new FlinkKafkaConsumer<>(ConfUtil.getTopicOfMall(), new SimpleStringSchema(), ConfUtil.getKafkaProperties()))
              .uid("kafka-source");

      // 进行分流处理
      SingleOutputStreamOperator<Object> mysqlBinlog = source
          .map(s -> JSON.parseObject(s, MysqlBinlog.class)).uid("map-mysqlbinlog")
          .process(new ProcessFunction<MysqlBinlog, Object>() {
            @Override
            public void processElement(MysqlBinlog mysqlBinlog, Context context,
                Collector<Object> collector) throws Exception {
              if (MysqlDatabase.MALL.toString().equalsIgnoreCase(mysqlBinlog.getDatabase())) {
                if (MysqlTable.ORDER_GOODS.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                  context.output(orderGoodTag,
                      JSON.parseObject(mysqlBinlog.getData(), OrderGoods.class));
                }
              }
            }
          }).uid("process");

      // 获取订单测输出流
      DataStream<OrderGoods> orderGoods = mysqlBinlog.getSideOutput(orderGoodTag);

//      orderGoods.print();
      orderGoods
          // watermark
          .assignTimestampsAndWatermarks(
              new BoundedOutOfOrdernessTimestampExtractor<OrderGoods>(Time.seconds(5)) {
                @Override
                public long extractTimestamp(OrderGoods o) {
                  return o.getUpdated_time() > 0 ? o.getUpdated_time()*1000 : o.getCreated_time()*1000;
                }
              }
          )
          // 解决乱序将同一张表数据放在同一个分区，
          // 所有这边并行度要设置1，不然多个输入水位线取最小，会一直是long的最小值
          .setParallelism(1)
          // 过滤得到已退款和已付款的记录
          .filter(o->{
            Long zeroPointTimestamps = Util.getZeroPointTimestamps(o.getUpdated_time());
            return (o.getStatus() == 1 && o.getPaid_time() > zeroPointTimestamps)  || ((o.getStatus() == -3)&& o.getPaid_time() >zeroPointTimestamps);
          }).uid("filter")
          // 开窗，每日滚动，后面是时区偏移量(不然是8点到8点)
          .windowAll(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
//          .timeWindowAll(Time.days(1))
          // 窗口触发,1分钟触发一次
          .trigger(new TimerTrigger<OrderGoods>())
          // 处理函数
          .process(new OrderGoodsProcess())
              .uid("OrderGoodsProcess");

      env.execute("OrderAnalysis job");
    } catch (Exception e) {
//      e.printStackTrace();
      LOG.error("OrderAnalysis Running Error",e);
    }

  }
  private static class OrderGoodsProcess extends ProcessAllWindowFunction<OrderGoods,String,TimeWindow> {

    /**
     * 营业额
     */
    private ValueState<String> sumOfBusiness;
    /**
     * 订单id集合
     */
    private ValueState<Set> idState;
    private ValueState<Tuple4<BigDecimal,BigDecimal,BigDecimal,BigDecimal>> shopTypeTuple4;

    /**
     * 当前窗口的最大时间戳
     */
    private ValueState<Long> windowEnd;
    private String date;
    private SimpleDateFormat sdf;
    private JedisCluster jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
      super.open(parameters);
      idState = getRuntimeContext()
          .getState(new ValueStateDescriptor<Set>("orderIdState", Set.class,new HashSet<>()));

      shopTypeTuple4 = getRuntimeContext().getState(new ValueStateDescriptor<Tuple4<BigDecimal, BigDecimal, BigDecimal, BigDecimal>>(
              "shopTypeValue",TypeInformation.of(new TypeHint<Tuple4<BigDecimal, BigDecimal, BigDecimal, BigDecimal>>() {}), Tuple4.apply(new BigDecimal(0),new BigDecimal(0), new BigDecimal(0), new BigDecimal(0))));

      sumOfBusiness= getRuntimeContext().getState(new ValueStateDescriptor<String>("sumOfBusiness",String.class,"0"));
      windowEnd= getRuntimeContext().getState(new ValueStateDescriptor<Long>("windowEnd",Long.class,0L));
      sdf=new SimpleDateFormat("yyyyMMdd");
      jedis = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
    }

    @Override
    public void close() throws Exception {
      super.close();
      jedis.close();
    }

    @Override
    public void process(Context ctx, Iterable<OrderGoods> iterable,
        Collector<String> out) throws Exception {
      long end = ctx.window().maxTimestamp();
      Tuple4<BigDecimal, BigDecimal, BigDecimal, BigDecimal> value = shopTypeTuple4.value();
      BigDecimal bShop = value._1();
      BigDecimal cShop = value._2();
      BigDecimal factoryShop = value._3();
      BigDecimal happyFemaleCook = value._4();


      if(date==null){
        date=sdf.format(new Date(end));
      }
      if(windowEnd.value()!=end){
        //时间不一样就更新时间并清空状态
        windowEnd.update(end);
        idState.clear();
        sumOfBusiness.clear();
        date=sdf.format(new Date(end));
        //将数值赋值到pre上并清空当日shoptype状态
        shopTypeTuple4.clear();
        jedis.hset(ConfUtil.getShopTypePreKey(),"pre_1",bShop+"");
        jedis.hset(ConfUtil.getShopTypePreKey(),"pre_2",cShop+"");
        jedis.hset(ConfUtil.getShopTypePreKey(),"pre_3",factoryShop+"");
        jedis.hset(ConfUtil.getShopTypePreKey(),"Pre_4",happyFemaleCook+"");
        bShop = new BigDecimal(0);
        cShop = new BigDecimal(0);
        factoryShop = new BigDecimal(0);
        happyFemaleCook = new BigDecimal(0);
        //每日定时删除ttl过期时间的数据
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_YEAR,ConfUtil.getRedisTtlDay());
        String formatDate = sdf.format(calendar.getTime());
        for (String key:ConfUtil.getDelRedisKeys()){
          jedis.del(key+formatDate);
        }
      }
      Set<String> idSets = idState.value();
      String sum = sumOfBusiness.value();
      // 滚动每日营业额，订单数，另外清单显示排名前十的商家
      Iterator<OrderGoods> iterator = iterable.iterator();


      while(iterator.hasNext()){
        OrderGoods og = iterator.next();
        String fullId=og.getOrder_id()+":"+og.getId();
        String price ;
        if (og.getStatus()==1) { // 付款
          if(idSets.contains(fullId)){
            continue;
          }
          // 新增订单
          price=og.getPay_price();
          idSets.add(fullId);
        }else{//退款
          if (!idSets.contains(fullId)) {
            continue;
          }
          price="-"+og.getPay_price();
          idSets.remove(fullId);
        }
        String shopType = og.getShop_type();
        BigDecimal priceBigDecimal = new BigDecimal(price);
        //shoptype
        switch (shopType){
          case "1": bShop = bShop.add(priceBigDecimal);jedis.hset(ConfUtil.getShopTypePreKey(),shopType,bShop+"");break;
          case "2": cShop = cShop.add(priceBigDecimal);jedis.hset(ConfUtil.getShopTypePreKey(),shopType,cShop+"");break;
          case "3": factoryShop = factoryShop.add(priceBigDecimal);jedis.hset(ConfUtil.getShopTypePreKey(),shopType,factoryShop+"");break;
          case "4": happyFemaleCook = happyFemaleCook.add(priceBigDecimal);jedis.hset(ConfUtil.getShopTypePreKey(),shopType,happyFemaleCook+"");break;
          default:
        }
        shopTypeTuple4.update(new Tuple4<BigDecimal, BigDecimal, BigDecimal, BigDecimal>(bShop,cShop,factoryShop,happyFemaleCook));


        sum= MyMath.add(sum,price);
        // 新增退款,运费不退 负数
        jedis.zincrby(ConfUtil.getTopGoodsRedisPreKey()+date,Double.parseDouble(price),og.getGoods_id());
        jedis.zincrby(ConfUtil.getTopSellerRedisPreKey()+date,Double.parseDouble(price),og.getSeller_id());



      }
      // 更新状态
      sumOfBusiness.update(sum);
      idState.update(idSets);

      // 计算订单数
      Set<String> orderIdSet = new HashSet<>();
      idSets.forEach(o -> orderIdSet.add(o.substring(0,o.indexOf(":"))));

      jedis.set(ConfUtil.getSumNumRedisPreKey()+date, String.valueOf(orderIdSet.size()));
      jedis.set(ConfUtil.getSumMoneyRedisPreKey()+date, String.valueOf(sum));

    }
  }
}


